package com.mtiiot.mq.esdao;

import org.apache.commons.pool.BasePoolableObjectFactory;
import org.apache.commons.pool.impl.GenericObjectPool;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

import java.net.InetAddress;

/**
 * Created by Administrator on 16-7-31.
 */
public class ElasticsearchDataSource {
    protected int maxActive;
    protected int maxIdle;
    protected int minIdle;
    protected long maxWait;
    protected String ips; // hostname port
    protected String clusterName = "elasticsearch";

    public int getMaxActive() {
        return maxActive;
    }

    public void setMaxActive(int maxActive) {
        this.maxActive = maxActive;
    }

    public int getMaxIdle() {
        return maxIdle;
    }

    public void setMaxIdle(int maxIdle) {
        this.maxIdle = maxIdle;
    }

    public int getMinIdle() {
        return minIdle;
    }

    public void setMinIdle(int minIdle) {
        this.minIdle = minIdle;
    }

    public long getMaxWait() {
        return maxWait;
    }

    public void setMaxWait(long maxWait) {
        this.maxWait = maxWait;
    }

    public String getIps() {
        return ips;
    }

    public void setIps(String ips) {
        this.ips = ips;
    }

    public String getClusterName() {
        return clusterName;
    }

    public void setClusterName(String clusterName) {
        this.clusterName = clusterName;
    }

    private GenericObjectPool pool;

    public void initPool()throws Exception{
        GenericObjectPool.Config config=new GenericObjectPool.Config();
        config.maxActive=this.maxActive;
        config.maxIdle=maxIdle;
        config.minIdle=minIdle;
        config.maxWait=maxWait;
        ClientFactory factory = new ClientFactory(this.ips,this.clusterName);
        pool = new GenericObjectPool(factory, config);
    }

    public Client getClient() throws Exception{
        if(pool==null){
            initPool();
        }
        return (Client)pool.borrowObject();
    }

    public void releaseClient(Client client)throws Exception{
        if(pool==null){
            initPool();
        }
        pool.returnObject(client);
    }


    /**
     * inner
     * @author qing
     *
     */
    class ClientFactory extends BasePoolableObjectFactory {
        private String innerIps;
        private String innerClusterName;
        public ClientFactory(String ips,String clusterName){
            innerIps=ips;
            innerClusterName=clusterName;
        }
        @Override
        public Object makeObject() throws Exception {
            Settings settings = Settings.settingsBuilder()
                    .put("client.transport.sniff", true)
                    .put("cluster.name", innerClusterName).build();
            TransportClient tc=TransportClient.builder().settings(settings).build();
            String[] tmp1=innerIps.split(",");
            for (String str : tmp1) {
                String host=str.split(":")[0];
                int port=Integer.parseInt(str.split(":")[1]);
                tc=tc.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), port));
            }
            Client client = tc;
            return client;
        }

        public void destroyObject(Object obj) throws Exception  {
            if(obj instanceof Client){
                Client client=(Client)obj;
                if(client!=null){
                    client.close();
                }
            }
        }
    }
}
